package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author Felix
 * @date 2023/7/29
 * 将流中的数据写到hbase表中
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //{"tm_name":"苹果111","id":2,"type":"update",
        // "table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info",
        // "sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
        //获取操作类型
        String type = jsonObj.getString("type");
        //获取当前这条数据对应的配置信息
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);
        String sinkTable = tableProcess.getSinkTable();
        String rowKey = jsonObj.getString(tableProcess.getSinkRowKey());
        //判断对业务数据库的维度表做了什么操作
        if ("delete".equals(type)) {
            //从hbase数据库表中将对应的数据删除掉
            System.out.println("~~从hbase中删除" + sinkTable + "表的" + rowKey + "数据~~");
            HbaseUtil.delRow(conn, GmallConfig.HBASE_NAMESPACE, sinkTable, rowKey);
        } else {
            //如果对维度表做了update或者insert操作    将维度数据put到hbase表中
            System.out.println("~~向hbase的" + sinkTable + "中put" + rowKey + "数据~~");
            String sinkFamily = tableProcess.getSinkFamily();
            String[] columns = tableProcess.getSinkColumns().split(",");
            String[] values = new String[columns.length];
            for (int i = 0; i < values.length; i++) {
                values[i] = jsonObj.getString(columns[i]);
            }
            HbaseUtil.putRow(conn,GmallConfig.HBASE_NAMESPACE,sinkTable,rowKey,sinkFamily,columns,values);
        }
        //如果维度数据发生了变化，将Redis中缓存的维度数据删除掉
        if("delete".equals(type)||"update".equals(type)){
            DimUtil.delCached(sinkTable,rowKey);
        }
    }
}
